package com.example.testi18n.service.impl;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.example.testi18n.entity.RedEnvelopeCopy1;
import com.example.testi18n.entity.RedEnvelopeCopy2;
import com.example.testi18n.exception.UserServiceException;
import com.example.testi18n.exception.constant.UserThrowException;
import com.example.testi18n.service.RedEnvelopeCopy1Service;
import com.example.testi18n.service.RedEnvelopeCopy2Service;
import com.example.testi18n.service.SyncService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Service
public class SyncServiceImpl implements SyncService {

    @Resource
    private TaskExecutor taskExecutor;

    @Resource
    private RedEnvelopeCopy1Service redEnvelopeCopy1Service;

    @Resource
    private RedEnvelopeCopy2Service redEnvelopeCopy2Service;

    /**
     * 一次同步多少条数据(分页大小)
     */
    private final int batchSize = 1000;
    /**
     * 每批次执行多少线程去执行（默认5）
     */
    private final int threadCount = 10;

    /**
     * 多线程同步数据
     * 1、根据待处理的数据总量，计算总共分成多少批去执行（批次总数）
     * 2、根据批次总数和默认线程数量，计算拆分成几个Task（任务总数）
     * 3、每个Task默认5个线程去执行，若当前Task剩余待处理批次不满足五个，将批次总数对最大5线程数取余，
     *   即为当前任务下的线程数量（每个任务的线程数,最多5个）
     */
    @Override
    public void testAsyncData() {
        try {
            /** 1、查询需要处理的数据量 **/
            int dataCount = redEnvelopeCopy1Service.count();// 数据总量
            log.info("待处理数据总量:{}", dataCount);
            if (dataCount < 1) {
                log.error("没有需要同步的数据, 结束!");
                return;
            }

            /** 2、批次数:当前需要执行多少次(数据量/批次大小) **/
            int execNext = 0;

            if (dataCount % batchSize == 0){
                execNext = dataCount / batchSize;
            }else {
                execNext = (dataCount / batchSize) + 1;
            }

            /** 3、并发线程下执行的任务批次(批次数/默认线程数) **/
            int taskNext = 0;
            if (execNext % threadCount == 0){
                taskNext = execNext / threadCount;
            }else {
                taskNext = (execNext / threadCount) + 1;
            }
            log.info("start------开始执行同步数据，数据总量: {}, 每次处理数据量：{}， 需要处理次数: {}。 单次并发线程数: {}, 执行批次：{}",
                    dataCount, batchSize, execNext, threadCount, taskNext);

            /** 4、页数 **/
            int pageIndex = 1;
            for (int i = 0; i < taskNext; i++) {
                // 本次需要的线程数，默认五个，
                // 如果当前执行次数 * 并发线程数 > 总共执行的次数，说明当前这次不够threadCount的数量，取余数就可以
                int threadNumber = threadCount;
                if ((i + 1) * threadCount > execNext) {
                    threadNumber = execNext % threadCount;
                }
                log.info("当前执行次数：{}, 本次执行线程数：{}", i + 1, threadNumber);

                CountDownLatch count = new CountDownLatch(threadNumber);
                for (int j = 0; j < threadNumber; j++) {
                    AtomicInteger currPage = new AtomicInteger(pageIndex);
                    taskExecutor.execute(() ->{
                        try {
                            long startTime = System.currentTimeMillis();
                            /** 4、查询数据 **/
                            List<RedEnvelopeCopy1> redEnvelopeCopy1s = this.selectList(currPage.get());
                            /** 5、处理数据 **/
                            this.dealWithData(redEnvelopeCopy1s);
                            log.info("同步成功：第 {} 页, 数据量: {}条, 耗时:{}秒", currPage.get(), redEnvelopeCopy1s.size(), (System.currentTimeMillis() - startTime) / 1000);
                        } finally {
                            count.countDown();
                        }
                    });
                    pageIndex++;
                }
                // 阻塞当前执行任务中的所有线程，等待本次线程全部执行完再开启下一次
                count.await();
            }
        } catch (Exception e) {
            log.error("同步数据错误,error content: {}", e.getMessage());
            throw new UserServiceException(UserThrowException.SERVER_ERROR);
        }

    }


    /**
     * 查询数据(分页)
     * @param pageNumber 当前页数
     */
    public List<RedEnvelopeCopy1> selectList(int pageNumber) {
        List<RedEnvelopeCopy1> sourceList = new ArrayList<>();
        // batchSize 为页大小
        Page<Object> page = new Page<>();
        page.setSize(batchSize);
        page.setCurrent(pageNumber);
        IPage<RedEnvelopeCopy1> redEnvelopeCopy1IPage = redEnvelopeCopy1Service.listByPage(page);
        if (!ObjectUtils.isEmpty(redEnvelopeCopy1IPage)){
            sourceList = redEnvelopeCopy1IPage.getRecords();
        }

        return sourceList;
    }

    /**
     * 处理数据
     * @param sourceList 待处理数据
     */
    @Transactional(rollbackFor = Exception.class)
    public void dealWithData(List<RedEnvelopeCopy1> sourceList) {
        log.info("......开始处理数据......");
        List<RedEnvelopeCopy2> list = new ArrayList<>();

        /** 此处批量添加有问题 **/
        for (RedEnvelopeCopy1 redEnvelopeCopy1 : sourceList) {
            RedEnvelopeCopy2 redEnvelopeCopy2 = new RedEnvelopeCopy2();
            BeanUtils.copyProperties(redEnvelopeCopy1, redEnvelopeCopy2);
            list.add(redEnvelopeCopy2);
        }

        /** 批量保存 **/
        redEnvelopeCopy2Service.saveBatch(list);

    }
}
